package com.kitten.dataAlign.consumer;

import com.kitten.dataAlign.constants.MQConstants;
import com.kitten.dataAlign.constants.RedisKeyConstants;
import com.kitten.dataAlign.constants.TableConstants;
import com.kitten.dataAlign.domain.mapper.InsertRecordMapper;
import com.kitten.dataAlign.model.dto.FollowUnfollowMqDTO;
import com.kitten.framework.common.utils.JsonUtils;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.Objects;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_data_align_" + MQConstants.TOPIC_COUNT_FOLLOWING,
    topic = MQConstants.TOPIC_COUNT_FOLLOWING
)
public class TodayUserFollowingCountIncrementData2DBConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private InsertRecordMapper insertRecordMapper;

    @Value("${table.shards}")
    private int tableShards;

    @Override
    public void onMessage(String body) {
        log.info("用户关注/取关 日增量数据变更 消费");
        FollowUnfollowMqDTO dto = JsonUtils.parseObject(body, FollowUnfollowMqDTO.class);

        Long userId = dto.getUserId();  // 用户ID
        Long targetUserId = dto.getTargetUserId();  // 被关注/取关的用户ID

        String date = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));  // 今日日期字符串

        // ---------------------- 用户关注数记录 --------------------------
        String userBloomKey = RedisKeyConstants.buildBloomUserFollowListKey(date);  //bloom:dataAlign:user:follows:
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setResultType(Long.class);
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/followUnfollow/bloom_today_user_follow_check.lua")));
        Long result = redisTemplate.execute(script, Collections.singletonList(userBloomKey), userId);

        RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
        if (Objects.equals(result, 0L)) {
            // 若bloom中没有记录, 则落库, 减轻DB压力
            long userIdHashKey = userId % tableShards;

            try {
                // 将日增量变更数据，写入表 t_data_align_following_count_temp_日期_分片序号
                insertRecordMapper.insert2DataAlignUserFollowTempTable(
                        TableConstants.buildTableNameSuffix(date, userIdHashKey), userId);
            } catch (Exception e) {
                log.error("", e);
            }
            // 记录用户ID到bloom中
            redisTemplate.execute(bloomAddScript, Collections.singletonList(userBloomKey), userId);

        }
        // ---------------------- 用户粉丝数记录 --------------------------
        String targetUserBloomKey = RedisKeyConstants.buildBloomUserFansListKey(date);  //bloom:dataAlign:user:fans:
        result = redisTemplate.execute(script, Collections.singletonList(targetUserBloomKey), targetUserId);
        if (Objects.equals(result, 0L)) {
            // 若bloom中没有记录, 则落库, 减轻DB压力
            long targetUserIdHashKey = targetUserId % tableShards;

            try {
                // 将日增量变更数据，写入表 t_data_align_fans_count_temp_日期_分片序号
                insertRecordMapper.insert2DataAlignUserFansTempTable(
                        TableConstants.buildTableNameSuffix(date, targetUserIdHashKey), targetUserId);
            } catch (Exception e) {
                log.error("", e);
            }
            // 记录用户ID到bloom中
            redisTemplate.execute(bloomAddScript, Collections.singletonList(targetUserBloomKey), targetUserId);
        }


    }
}
